package org.jrazdacha.bittorrent.modelling.agent;

import jade.core.AID;
import jade.core.Agent;
import jade.core.behaviours.CyclicBehaviour;
import jade.core.behaviours.OneShotBehaviour;
import jade.core.behaviours.TickerBehaviour;
import jade.lang.acl.ACLMessage;
import jade.lang.acl.UnreadableException;

import java.io.IOException;
import java.util.BitSet;
import java.util.Set;

import org.jrazdacha.bittorrent.modelling.core.Communicator;
import org.jrazdacha.bittorrent.modelling.core.ConversationId;
import org.jrazdacha.bittorrent.modelling.core.Id;
import org.jrazdacha.bittorrent.modelling.core.Message;
import org.jrazdacha.bittorrent.modelling.core.Peer;
import org.jrazdacha.bittorrent.modelling.core.Torrent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

/**
 * Agent which represents bittorrent peer.
 * It registers in DF(Directory Facilitator) which represents bittorrent tracker.
 * After that peer agent consumes other registered peers from DF. 
 * 
 * @author Vitaly Ruzhnikov
 *
 */
public class PeerAgent extends Agent implements Communicator {
	private static final int INTERVAL = 30000;
	final int PIECE_SIZE = 2048;
	final int GB_40 = 335544320;

	private static final long serialVersionUID = 3528475905220266751L;
	
	private static final String AGENT_ID = "agentid";	
	private static final Logger logger = LoggerFactory.getLogger(PeerAgent.class);
	private Peer peer;	

	@Override
	protected void setup() {		
		putAgentIdForLogging();		
		register();        
		setupPeer();
        addAnnounceBehaviour();
        
        addBehaviour(new CyclicBehaviour() {
			
			@Override
			public void action() {
				logger.debug("Try to receive messages");
				ACLMessage msg = receive();				
			
				if (msg != null) {					
					Message message = null;
					logger.debug("Message received. Conversation id is {} from", msg.getConversationId(), msg.getSender());
					if (ConversationId.HANDSHAKE.toString().equals(msg.getConversationId())) {
						logger.debug("Handshake received", msg.getConversationId());
						message = new Message(ConversationId.HANDSHAKE);										
					} else if (ConversationId.BITFIELD.toString().equals(msg.getConversationId())) {
						logger.debug("Bitfield received", msg.getConversationId());
						try {
							message = new Message(ConversationId.BITFIELD, (BitSet)msg.getContentObject());
						} catch (UnreadableException e) {
							throw new RuntimeException(e);
						}
					}
					peer.receiveMessageFrom(message, new IdAIDImpl(msg.getSender()));
					logger.debug("Connection state with remote peer {} is {}", msg.getSender(), peer.getConnectionTo(new IdAIDImpl(msg.getSender())).getState());
				} else {
					block();
				}
			}
		});
        
		super.setup();
	}

	private void setupPeer() {
		peer = new Peer(new IdAIDImpl(getAID()), this);	
		peer.startLeeching(new Torrent(GB_40, PIECE_SIZE));
	}
	
	private void putAgentIdForLogging() {
		MDC.put(AGENT_ID, getAID().getName());
	}	

	private void register() {
		DFUtils.registerAgent(this);
	}

	private void addAnnounceBehaviour() {
		addBehaviour(new TickerBehaviour(this, INTERVAL) {					
			private static final long serialVersionUID = 746571748445518849L;

			@Override
			protected void onTick() {						
				logger.debug("Ask tracker for more peers");
				Set<Id> peerIds = askPeers();
				logger.debug("Peers received : " + peerIds.size());							
				for (Id peerId : peerIds) {
					logger.debug("Id {}", ((AID)peerId.getValue()).getName());
				}
				peer.connectTo(peerIds);
			}
		});
	}		
	
	private Set<Id> askPeers() {		
		return DFUtils.askPeers(this);
	}

	@Override
	protected void takeDown() {
		DFUtils.deregisterAgent(this);
		super.takeDown();
	}

	@Override
	public void sendHandShakeTo(final Id remotePeerId) {
		addBehaviour(new OneShotBehaviour() {
			
			@Override
			public void action() {
				logger.debug("Connection state with remote peer {} is {}", remotePeerId, peer.getConnectionTo(remotePeerId).getState());
				ACLMessage message = new ACLMessage(ACLMessage.CFP);
				message.setConversationId(ConversationId.HANDSHAKE.toString());
				message.addReceiver((AID)remotePeerId.getValue());
				send(message);
				logger.debug("Send handshake to {}", remotePeerId.getValue());
				logger.debug("Connection state with remote peer {} is {}", remotePeerId, peer.getConnectionTo(remotePeerId).getState());
			}
		});			
	}

	@Override
	public void sendBitfieldTo(final BitSet bitfield, final Id remotePeerId) {
		addBehaviour(new OneShotBehaviour() {
			@Override
			public void action() {
				logger.debug("Connection state with remote peer {} is {}", remotePeerId, peer.getConnectionTo(remotePeerId).getState());
				ACLMessage message = new ACLMessage(ACLMessage.INFORM);
				message.setConversationId(ConversationId.BITFIELD.toString());
				message.addReceiver((AID)remotePeerId.getValue());
				try {
					message.setContentObject(bitfield);
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
				send(message);
				logger.debug("Send bitfield to {}", remotePeerId.getValue());
				logger.debug("Connection state with remote peer {} is {}", remotePeerId, peer.getConnectionTo(remotePeerId).getState());
			}
		});		
	}
	

}
